package mya.mes.mdc.weixin;

import mya.mes.mdc.share.vo.MdcOptions;
import org.fusesource.hawtbuf.Buffer;
import org.fusesource.hawtbuf.UTF8Buffer;
import org.fusesource.mqtt.client.*;

import java.io.UnsupportedEncodingException;
import java.net.URISyntaxException;

public class WxMqttClient {

    private MdcOptions mdcOptions;
    private MQTT mqtt;

    private UTF8Buffer topic;
    private BlockingConnection blockingConnection;
    private CallbackConnection connection;
    private volatile boolean isConnecting = false;

    void init(MdcOptions mdcOptions) {
        this.mdcOptions = mdcOptions;
        // this.msgReceiver = new MsgReceiver(mdcOptions);
        mqtt = new MQTT();
        try {
            mqtt.setHost(mdcOptions.WEIXIN_MQTT_IP, mdcOptions.WEIXIN_MQTT_PORT);
        } catch (URISyntaxException urs) {
            urs.printStackTrace();
        }
        mqtt.setClientId(mdcOptions.WEIXIN_MQTT_CLIENT_ID);
        mqtt.setCleanSession(true);
        mqtt.setKeepAlive((short)10);
//        mqtt.setConnectAttemptsMax(10L);//客户端首次连接到服务器时，连接的最大重试次数，超出该次数客户端将返回错误。-1意为无重试上限，默认为-1
//        mqtt.setReconnectAttemptsMax(2L);//客户端已经连接到服务器，但因某种原因连接断开时的最大重试次数，超出该次数客户端将返回错误。-1意为无重试上限，默认为-1
//        mqtt.setReconnectDelay(1000L);//首次重连接间隔毫秒数，默认为10ms
//        mqtt.setReconnectDelayMax(3000L);//重连接间隔毫秒数，默认为30000ms
//        mqtt.setReconnectBackOffMultiplier(2);

        mqtt.setVersion("3.1.1");
        topic = UTF8Buffer.utf8(mdcOptions.WEIXIN_MQTT_TOPIC);
//        blockingConnection = mqtt.blockingConnection();
        connection = mqtt.callbackConnection();

    }

    public void publish(String content) throws Exception {
        connection.listener(new Listener() {
            @Override
            public void onConnected() {
                System.out.println("remote MQTT onConnected");
            }

            @Override
            public void onDisconnected() {
                System.out.println("remote MQTT onDisconnected");
                disconnect();
                isConnecting = false;
                connection =mqtt.callbackConnection();
            }

            @Override
            public void onPublish(UTF8Buffer topic, Buffer body, Runnable ack) {

            }

            @Override
            public void onFailure(Throwable value) {
                System.out.println("remote MQTT onFailure");
            }
        });
        if(isConnecting == false){
            connection.connect(new Callback<Void>() {
                public void onFailure(Throwable ex) {
                    System.out.println("remote connect onFailure");
                    isConnecting = false;
                }
                public void onSuccess(Void v) {
                    try {
                        isConnecting = true;
                        connection.publish(mdcOptions.WEIXIN_MQTT_TOPIC, content.getBytes("UTF-8"), QoS.AT_LEAST_ONCE, false, new Callback<Void>() {
                            public void onSuccess(Void v) {
                                System.out.println("remote content:"+content);
                                System.out.println("remote publish onSuccess");
                            }
                            public void onFailure(Throwable value) {
                                System.out.println("remote publish onFailure");

                            }
                        });
                    } catch (UnsupportedEncodingException e) {
                            e.printStackTrace();
                    }
                }
            });
        }else{
            connection.publish(mdcOptions.WEIXIN_MQTT_TOPIC, content.getBytes("UTF-8"), QoS.AT_LEAST_ONCE, false, new Callback<Void>() {
                public void onSuccess(Void v) {
                    System.out.println("remote content:"+content);
                    System.out.println("remote publish onSuccess");
                }
                public void onFailure(Throwable value) {
                    System.out.println("remote publish onFailure");
                }
            });
        }

//        if (!blockingConnection.isConnected()) {
//            System.out.println(" blockingConnection.connect ");
//            try {
//                blockingConnection.connect();
//            }catch (Exception e){
//                System.out.println("blockingConnection.connect 异常------------------------------");
//            }
//
//        }
//        if (topic == null) {
//            topic = UTF8Buffer.utf8(mdcOptions.WEIXIN_MQTT_TOPIC);
//        }
//        System.out.println("-------start publish---------------------------------------------");
//        blockingConnection.publish(topic,
//                UTF8Buffer.utf8(content),
//                QoS.AT_LEAST_ONCE, true);
//        System.out.println("----------------------------end publish---------");
    }


    public void shutdown() throws Exception {
        if (blockingConnection == null) {
            return;
        }
        if (blockingConnection.isConnected()) {
            blockingConnection.disconnect();
        }
        blockingConnection.kill();
    }

    public void disconnect() {
        connection.disconnect(new Callback<Void>() {
            public void onSuccess(Void v) {

            }
            public void onFailure(Throwable value) {

            }
        });
    }
}
